package com.bigdata.kafka;


import org.apache.commons.cli.*;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;

/**
 * kafka 数据文件发送工具
 *
 * @author xin.ding
 * @date 2020/8/18 14:12
 */
public class SendFile {
    private static final Logger logger = LoggerFactory.getLogger(SendFile.class);
    private static String bootstrapServer = null;
    private static String topic = null;
    private static String file = null;
    private static int delayTime = 0;
    private static String key = null;
    private static String saslUsername = null;
    private static String saslPassword = null;
    private static String krb5Conf = null;
    private static String jaasConf = null;
    private static boolean isHelp = false;
    private static boolean isDebug = false;
    private static boolean isPrint = false;
    private static final FastDateFormat DATE_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss.SSS");

    public static void main(String[] args) throws Exception {
        parseArgs(args);


        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServer);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("linger.ms", "5");
        props.put("acks", "all");

        if (saslUsername != null && saslPassword != null) {
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            props.put("sasl.jaas.config",
                    String.format("org.apache.kafka.common.security.plain.PlainLoginModule required username=\"%s\" password=\"%s\";"
                            , saslUsername
                            , saslPassword));
        }

        try (Producer<String, String> producer = new KafkaProducer<>(props);
             BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                ProducerRecord<String, String> msg;
                if (null != key) {
                    msg = new ProducerRecord<>(topic, key, line);
                } else {
                    msg = new ProducerRecord<>(topic, line);
                }
                producer.send(msg);
                if (isPrint) {
                    logger.info("[{}] Send: {}", DATE_FORMAT.format(System.currentTimeMillis()), line);
                }
                if (delayTime > 0) {
                    Thread.sleep(delayTime);
                }
            }
        } catch (IOException e) {
            logger.error("data send error", e);
        }
    }

    private static void parseArgs(String[] args) throws ParseException {
        CommandLineParser parser = new DefaultParser();
        Options options = new Options();

        options.addOption("bs", "bootstrap-server", true, "bootstrap-server");
        options.addOption("t", "topic", true, "topic");
        options.addOption("f", "file", true, "data file");
        options.addOption("", "delay", true, "delay time");
        options.addOption("pm", "print-message", false, "print message");
        options.addOption("k", "key", true, "message key");
        options.addOption("su", "sasl-username", true, "sasl-username");
        options.addOption("sp", "sasl-password", true, "sasl-password");
        options.addOption("kf", "krb5-conf", true, "krb5-conf");
        options.addOption("jf", "jaas-conf", true, "jaas-conf");
        options.addOption("d", "debug", false, "debug");
        options.addOption("h", "help", false, "help info");

        CommandLine commandLine = parser.parse(options, args, true);

        if (commandLine.hasOption("bootstrap-server") || commandLine.hasOption("bs")) {
            bootstrapServer = commandLine.getOptionValue("bs") != null ? commandLine.getOptionValue("bs") : commandLine.getOptionValue("bootstrap-server");
        }
        if (commandLine.hasOption("topic")) {
            topic = commandLine.getOptionValue("topic");
        }
        if (commandLine.hasOption("file") || commandLine.hasOption("f")) {
            file = commandLine.getOptionValue("f") != null ? commandLine.getOptionValue("f") : commandLine.getOptionValue("file");
        }
        if (commandLine.hasOption("delay")) {
            delayTime = Integer.parseInt(commandLine.getOptionValue("topic").trim());
        }
        if (commandLine.hasOption("print-message") || commandLine.hasOption("pm")) {
            isPrint = true;
        }
        if (commandLine.hasOption("key")) {
            key = commandLine.getOptionValue("key");
        }
        if (commandLine.hasOption("sasl-username") || commandLine.hasOption("su")) {
            saslUsername = commandLine.getOptionValue("su") != null ? commandLine.getOptionValue("su") : commandLine.getOptionValue("sasl-username");
        }
        if (commandLine.hasOption("sasl-password") || commandLine.hasOption("sp")) {
            saslPassword = commandLine.getOptionValue("sp") != null ? commandLine.getOptionValue("sp") : commandLine.getOptionValue("sasl-password");
        }
        if (commandLine.hasOption("krb5-conf") || commandLine.hasOption("kf")) {
            krb5Conf = commandLine.getOptionValue("kf") != null ? commandLine.getOptionValue("kf") : commandLine.getOptionValue("krb5-conf");
        }
        if (commandLine.hasOption("jaas-conf") || commandLine.hasOption("jf")) {
            jaasConf = commandLine.getOptionValue("jf") != null ? commandLine.getOptionValue("jf") : commandLine.getOptionValue("jaas-conf");
        }
        if (commandLine.hasOption("debug") || commandLine.hasOption('d')) {
            isDebug = true;
            delayTime = 1000;
            isPrint = true;
        }
        if (commandLine.hasOption("help") || commandLine.hasOption('h')) {
            isHelp = true;
        }
        if (args.length == 0 || isHelp) {
            String helpInfo = "\t  Option\t\t\t\tDescription\n" +
                    "\t--bootstrap-server[bs]\tbootstrap-server\n" +
                    "\t--topic\t\t\t\t\ttopic\n" +
                    "\t--file\t\t\t\t\tdata file\n" +
                    "\t--delay\t\t\t\t\tdelay time\n" +
                    "\t--print\t\t\t\t\tprint message\n" +
                    "\t--key\t\t\t\t\tmessage key\n" +
                    "\t--sasl-username[su]\t\tsasl username\n" +
                    "\t--sasl-password[sp]\t\tsasl password\n" +
                    "\t--krb5-conf[kf]\t\t\tkrb5-conf\n" +
                    "\t--jaas-conf[jf]\t\t\tjaas-conf\n" +
                    "\t--interval[i]\t\t\ttimeinterval\n" +
                    "\t--help[h]\t\t\t\thelp info\n";
            logger.info(helpInfo);
            System.exit(0);
        }

    }
}
